package com.wulaobo.state;

import com.wulaobo.constant.Constants;
import com.wulaobo.functions.WaterSensorMapFunction;
import com.wulaobo.pojo.WaterSensor;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author wuwenbo
 * @Date 2024/7/10 11:49
 * @ClassName: KeyedMapStateDemo
 * @Description: 案例需求：统计每种传感器每种水位值出现的次数。
 */
public class KeyedMapStateDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream(Constants.IP, 7777)
                .map(new WaterSensorMapFunction());

        sensorDS.keyBy(WaterSensor::getId).process(new KeyedProcessFunction<String, WaterSensor, String>() {

            MapState<Integer,Integer> mapState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);

                // TODO 2.在open方法中，初始化状态
                // 状态描述器两个参数：第一个参数，起个名字，不重复；第二个参数，存储的类型
                mapState = getRuntimeContext().getMapState(new MapStateDescriptor("mapState", Types.INT,Types.INT));
            }

            @Override
            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                if(mapState.contains(value.getVc())){
                    mapState.put(value.getVc(), mapState.get(value.getVc())+1);
                }else{
                    mapState.put(value.getVc(),1);
                }
                out.collect("key:"+value.getId()+",vc:"+value.getVc()+",出现次数是："+mapState.get(value.getVc()));
            }

        }).print();

        env.execute();

    }

}
